package com.bd08.flink.demo.zhang.table

import com.bd08.flink.demo.jing.table.Demo2.stenv
import org.apache.flink.api.scala.{ExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.Expressions.{$, row}
import org.apache.flink.table.api.{DataTypes, Schema}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object Demo2 extends App {
  //创建批执行上下文
  //val bevn = ExecutionEnvironment.getExecutionEnvironment

  //创建流执行上下文
  val env = StreamExecutionEnvironment.getExecutionEnvironment
  //创建流表执行上下文
  val stenv = StreamTableEnvironment.create(env)
  //变化Csv中数据为row注册成表
  val ds = env.readTextFile("H:\\Module 14 - 项目部署\\demo1\\src\\main\\resources\\zhang\\customers.csv")
 val rsds =  ds
    .map(line =>{
      val fileds = line.split("[\",]{1,3}")
      val rs=  fileds.toList.tail
      (rs(0),rs(1),rs(2),rs(3),rs(4),rs(5),rs(6),rs(7),rs(8))
    })

 // stenv.createTemporaryView("customer",rsds)
  val schame = Schema.newBuilder()
    .fromRowDataType(DataTypes.ROW(
      DataTypes.FIELD("name1",DataTypes.STRING().notNull()),
      DataTypes.FIELD("age2",DataTypes.STRING().notNull()),
      DataTypes.FIELD("birth3",DataTypes.STRING().notNull()),
      DataTypes.FIELD("birth4",DataTypes.STRING().notNull()),
      DataTypes.FIELD("birth5",DataTypes.STRING().notNull()),
      DataTypes.FIELD("birth6",DataTypes.STRING().notNull()),
      DataTypes.FIELD("birth7",DataTypes.STRING().notNull()),
      DataTypes.FIELD("birth8",DataTypes.STRING().notNull()),
      DataTypes.FIELD("birth9",DataTypes.STRING().notNull())
    )).build()
  stenv.createTemporaryView("customer",rsds)


  // stenv.executeSql("CREATE TABLE customerWhitScheam(id String,name string,name7 string,name6 string,name5 string,name4 string,name3 string,name2 string,name1 string) with ('connector' = 'filesystem','path' =  'H:\\Module 14 - 项目部署\\demo1\\src\\main\\resources\\zhang\\cum','format' = 'file')")
  //起 列别名
  val ct = stenv.from("customer").as("name9","name8","name7","name6","name5","name4","name3","name2","name1")
  ct.select($("*")).limit(5).execute().print()
  //  ct.executeInsert("customerWhitScheam")
  //  stenv.executeSql("select * from customerWhitScheam limit 5")

}
